package ex.datastream.physicalPartition;

import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class PhysicalPartitionTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);

        DataStream<String> dataStream = env.fromElements("hello", "world", "flink", "1", "2", "3", "aa", "bb", "cc", "11", "21", "33");
//        DataStream<String> broadcast = dataStream.broadcast();
//每个分区都发一遍
//        DataStream<String> broadcast = dataStream.rescale();

//        DataStream<String> broadcast = dataStream.rebalance();

        DataStream<String> broadcast = dataStream.shuffle();
        broadcast.print();
//        DataStream<String> broadcast = dataStream.global();
// 并行度为1，一次发
//        customPartition(env);
        env.execute();
    }

    private static void customPartition(StreamExecutionEnvironment env) {
        // 将自然数按照奇偶分区
        DataStream<Integer> dataStream = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8);
        dataStream.partitionCustom(new Partitioner<Integer>() {
            @Override
            public int partition(Integer key, int numPartitions) {
                return key % 2;
            }
        }, new KeySelector<Integer, Integer>() {
            @Override
            public Integer getKey(Integer value) throws Exception {
                return value;
            }
        }).print().setParallelism(2);
    }
//2> 1
//2> 3
//2> 5
//2> 7
//1> 2
//1> 4
//1> 6
//1> 8
}
